package com.atguigu.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

//窗口操作
object SparkStreaming06_Window {
//  使用SparkStreaming 完成WordCount
  def main(args: Array[String]): Unit = {

      val ints = List(1,2,3,4,5,6)
//    sliding(3,3)第一个三是窗口的大小范围，第二个三时窗口的步长
    val intses = ints.sliding(3,3)
//    窗口的大小
//    val intses = ints.sliding(3)
    for (elem <- intses) {
      println(elem)
    }
//    spark配置对象
     val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming01_WordCount")
//   实时数据分析环境对象
//   采集周期，以指定的时间为周期采集实时数据
     val streamingContext = new StreamingContext(sparkConf,Seconds(3))
//    保存的数据的状态，需要设定检查点路径
     streamingContext.sparkContext.setCheckpointDir("cp")
//    从kafka中采集数据
    val kafkaDStream:ReceiverInputDStream[(String,String)] = KafkaUtils.createStream(
      streamingContext,
      "hadoop102",
      //      以消费者组的概念消费数据
      "atguigu",
      //      Map("atguigu"->3)== atguigu是topic
      Map("atguigu" -> 3)
    )
//  窗口大小应该为采集周期的整数倍,窗口的滑动的步长也应该为采集周期的整数倍
    val windowDStream = kafkaDStream.window(Seconds(9),Seconds(3))
//   从指定文件夹中读取采集数据
//     val receiverDStream = streamingContext.receiverStream(new MyReceiver("hadoop102",9999))
//   将采集的数据进行分解（扁平化）
     val wordDStream = windowDStream.flatMap(t=>t._2.split(" "))
//   将数据进行结构的转换方便统计
     val mapDStream = wordDStream.map((_,1))
//
//
    val wordToSumDStream = mapDStream.reduceByKey(_+_)

    wordToSumDStream.print()
//    不能停止采集
//    streamingContext.stop()
//   启动采集器
    streamingContext.start()
//    Driver等待采集器的执行
    streamingContext.awaitTermination()
  }
}

